package com.dongzili.demo.customer.sync;

import cn.hutool.extra.spring.SpringUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.dongzili.demo.common.config.mybatisplus.MybatisPlusUtil;
import com.dongzili.demo.common.utils.ReflectUtils;
import com.dongzili.demo.customer.sync.mapper.CustomerMapper;
import lombok.extern.slf4j.Slf4j;
import org.dromara.easyes.core.kernel.BaseEsMapper;

import java.time.LocalDateTime;
import java.time.temporal.ChronoUnit;
import java.util.*;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
public class SyncService<T> {

    private final Class<T> tClass;
    private final String primaryColumnKey;
    private final String tableName;

    public SyncService(Class<T> tClass) {
        this.tClass = tClass;
        this.primaryColumnKey = ReflectUtils.getPrimaryColumnKey(tClass);
        this.tableName = ReflectUtils.getTableName(tClass);
    }

    public static <T> SyncService<T> getInstance(Class<T> tClass) {
        return new SyncService<>(tClass);
    }

    public int syncIndex(BaseEsMapper<T> baseEsMapper) {

        int size = 20000;
        LocalDateTime start = LocalDateTime.now();

        QueryWrapper<T> queryWrapper = new QueryWrapper<>();
        queryWrapper.orderByDesc(primaryColumnKey);
        Long total = MybatisPlusUtil.getMapper(tClass).selectCount(new LambdaQueryWrapper<>());
        int pages = (int) Math.ceil((double) total / size);
        log.info("同步索引开始，总共：{} 行，每页大小：{}，总页数：{}，耗时：{}秒", total, size, pages, ChronoUnit.SECONDS.between(start, LocalDateTime.now()));
        String first = SpringUtil.getBean(CustomerMapper.class).selectOne(tableName, primaryColumnKey, primaryColumnKey);
        if (first == null) {
            return size;
        }
        String firstId = first;
        List<String> ids = new ArrayList<>(Collections.singletonList(firstId));
        for(int i=1;i<=pages;i++){
            QueryWrapper<T> wrapper = new QueryWrapper<>();
            wrapper.select(primaryColumnKey);
            wrapper.orderByDesc(primaryColumnKey);
            wrapper.le(primaryColumnKey, firstId);
            wrapper.last("LIMIT 1 OFFSET "+size);
            T last = MybatisPlusUtil.getMapper(tClass).selectOne(wrapper);
            if (last != null) {
                String lastId = Objects.requireNonNull(ReflectUtils.getPrimaryValue(last)).toString();
                ids.add(lastId);
                firstId = lastId;
            }
        }
        AtomicInteger totalCount = new AtomicInteger(0);
        AtomicInteger index = new AtomicInteger(0);
        ids.stream().parallel().forEach(id -> {
            int i = index.getAndIncrement();
            LocalDateTime currentTime = LocalDateTime.now();
            QueryWrapper<T> wrapper = new QueryWrapper<>();
            wrapper.orderByDesc(primaryColumnKey);
            wrapper.le(primaryColumnKey, id);
            wrapper.last("LIMIT "+size);
            List<T> currentList = MybatisPlusUtil.getMapper(tClass).selectList(wrapper);
            Integer count = baseEsMapper.insertBatch(currentList);
//            RedisUtil.getHashService(tClass).addOrUpdateBatch(currentList);
            log.info("同步索引中...，当前页：{}，总页数:{}， 每页大小：{}，耗时：{}秒， 总耗时：{}秒", i, pages, size, ChronoUnit.SECONDS.between(currentTime, LocalDateTime.now()), ChronoUnit.SECONDS.between(start, LocalDateTime.now()));
            totalCount.addAndGet(currentList.size());
            // 手动置空，回收内存
            currentList = null;
            wrapper = null;
        });

//        log.info("同步索引结束，总共：{} 行，成功:{}行, 每页大小：{}, 总页数：{}，耗时：{}秒", total, size, totalCount.get(), size, ChronoUnit.SECONDS.between(start, LocalDateTime.now()));
        log.info("同步索引结束，总共：{} 行，成功:{}行, 每页大小：{}, 总页数：{}，耗时：{}秒", total, totalCount.get(), size, pages, ChronoUnit.SECONDS.between(start, LocalDateTime.now()));

        return totalCount.get();
    }

    //没有多线程
    public int syncIndex1(BaseEsMapper<T> baseEsMapper) {

        int size = 20000;
        LocalDateTime start = LocalDateTime.now();

        QueryWrapper<T> queryWrapper = new QueryWrapper<>();
        queryWrapper.orderByDesc(primaryColumnKey);
        AtomicInteger successCount = new AtomicInteger(0);
        List<T> list = new ArrayList<>(size);
        MybatisPlusUtil.getMapper(tClass).selectList(queryWrapper, resultContext -> {
            T t = resultContext.getResultObject();
            successCount.incrementAndGet();
            list.add(t);
            if(list.size() == size){
                baseEsMapper.insertBatch(list);
                list.clear();
                log.info("同步索引中...，当前页：{}, 总耗时：{}秒", resultContext.getResultCount(), ChronoUnit.SECONDS.between(start, LocalDateTime.now()));
            }
        });
        log.info("同步索引结束，成功:{}行, 每页大小：{}, 耗时：{}秒", successCount.get(), size, ChronoUnit.SECONDS.between(start, LocalDateTime.now()));
        return successCount.get();
    }

    public int syncIndex2(BaseEsMapper<T> baseEsMapper) {

        int size = 20000;
        LocalDateTime start = LocalDateTime.now();

        QueryWrapper<T> queryWrapper = new QueryWrapper<>();
        queryWrapper.select(primaryColumnKey);
        queryWrapper.orderByDesc(primaryColumnKey);
        queryWrapper.orderByDesc(primaryColumnKey);
        List<String> ids = new ArrayList<>();
        int pages = 1;
        AtomicInteger totalCount = new AtomicInteger(0);
        MybatisPlusUtil.getMapper(tClass).selectList(queryWrapper, resultContext -> {
            totalCount.incrementAndGet();
            T t = resultContext.getResultObject();
            if(resultContext.getResultCount() % size == 1){
                ids.add(Objects.requireNonNull(ReflectUtils.getPrimaryValue(t)).toString());
            }
        });
        log.info("查询索引id结束，总共：{} 行，耗时：{}秒", ids.size(), ChronoUnit.SECONDS.between(start, LocalDateTime.now()));
        AtomicInteger successCount = new AtomicInteger(0);
        AtomicInteger index = new AtomicInteger(0);
        ids.stream().parallel().forEach(id -> {
            int i = index.getAndIncrement();
            LocalDateTime currentTime = LocalDateTime.now();
            QueryWrapper<T> wrapper = new QueryWrapper<>();
            wrapper.orderByDesc(primaryColumnKey);
            wrapper.le(primaryColumnKey, id);
            wrapper.last("LIMIT "+size);
            List<T> currentList = MybatisPlusUtil.getMapper(tClass).selectList(wrapper);
            Integer count = baseEsMapper.insertBatch(currentList);
            log.info("同步索引中...，当前页：{}，总页数:{}， 每页大小：{}，耗时：{}秒， 总耗时：{}秒", i, pages, size, ChronoUnit.SECONDS.between(currentTime, LocalDateTime.now()), ChronoUnit.SECONDS.between(start, LocalDateTime.now()));
            totalCount.addAndGet(count);
            // 手动置空，回收内存
            currentList = null;
            wrapper = null;
        });

//        log.info("同步索引结束，总共：{} 行，成功:{}行, 每页大小：{}, 总页数：{}，耗时：{}秒", total, size, totalCount.get(), size, ChronoUnit.SECONDS.between(start, LocalDateTime.now()));
        log.info("同步索引结束，总共：{} 行，成功:{}行, 每页大小：{}, 总页数：{}，耗时：{}秒", totalCount.get(), successCount.get(), size, pages, ChronoUnit.SECONDS.between(start, LocalDateTime.now()));

        return totalCount.get();
    }


}
